﻿using Confluent.Kafka;
using Microsoft.AspNetCore.Mvc;
using Newtonsoft.Json.Linq;

namespace Demo.Net.API.Controllers
{

    [Route("api/[controller]/[action]")]
    [ApiController]
    public class KafkaControllers: ControllerBase
    {

    //    private readonly ProducerConfig _producerConfig;
        public KafkaControllers()
        { 

        }

        [HttpGet]
        public async Task<IActionResult> KafkaTest(string msg)
        {

            var _producerConfig = new ProducerConfig
            {
                BootstrapServers = "192.168.31.154:9092", // Kafka集群地址

               Acks = Acks.None,

              //  EnableIdempotence = true, //幂等性

                 
            //    ClientId = "your-client-id" // 客户端ID
            };

            _producerConfig.Acks=Acks.None;


            using (var producer = new ProducerBuilder<string, string>(_producerConfig).Build())
            {
                _producerConfig.ClientId = Guid.NewGuid().ToString();
                string topic = "topicName"; // 主题名称

               await  producer.ProduceAsync(topic, message: new Message<string, string>()
                {
                    Value =msg
                  
                });

            }

            return Ok();

        }

        [HttpGet]
        public async Task<IActionResult> KafkaTransTest(int a)
        {
          

            
                //开启事务 ack =all 
                var config = new ProducerConfig { BootstrapServers = "192.168.31.154:9092",Acks= Acks.All,TransactionalId= "2233233"};


                using (var producer = new ProducerBuilder<Null, string>(config).Build())
                {
                    // 初始化事务
                    producer.InitTransactions(TimeSpan.FromSeconds(10));

                    try
                    {
                        // 开始事务
                        producer.BeginTransaction();
                        
                        // 发送一条消息到Kafka
                        var message = new Message<Null, string> { Value = a.ToString() };
                        producer.Produce("topicName", message);

                        int b = 1;

                        int c;

                        c = b / a;

                        // 提交事务
                        producer.CommitTransaction();
                    }
                    catch (Exception ex)
                    {
                      
                        producer.AbortTransaction();

                    return Ok(ex.Message);
                    }
                }



            return Ok();
        }
    }
}
